#lang racket
(require racket/contract
		 "definition.rkt"
		 "utility.rkt"
		 "observable.rkt"
		 "subscriber.rkt"
		 "disposable.rkt")

(provide (except-out (all-defined-out) funn fune func))

(define-syntax (funn stx)
  (syntax-case stx ()
	[(_ sub) #'(lambda (x) (on-next sub x))]))

(define-syntax (fune stx)
  (syntax-case stx ()
	[(_ sub) #'(lambda (e) (on-error sub e))]))

(define-syntax (func stx)
  (syntax-case stx ()
	[(_ sub) #'(lambda () (on-completed sub))]))


(define/contract (rx/empty)
  (-> observable?)
  (create-observable (lambda (sub) (on-completed sub))))

(define/contract (rx/return value)
  (-> any/c observable?)
  (create-observable (lambda (sub)
					   (on-next sub value)
					   (on-completed sub))))

(define/contract (rx/interval second)
  (-> real? observable?)
  (create-observable
   (lambda (sub)
	 (thread (lambda ()
			   (define duration (* 1000 second))
			   (let loop ([now (current-inexact-milliseconds)])
				 (let ([x (+ now duration)])
				   (sync (alarm-evt x))
				   (on-next sub x)
				   (loop x))))))))


(define/contract (rx/never)
  (-> observable?)
  (create-observable void))

(define/contract (rx/merge* lst)
  (-> (listof observable?) observable?)
  (create-observable
   (lambda (sub)
	 (define count (box (length lst)))
	 (for-each
	  (lambda (obs)
		(subscribe obs (funn sub) (fune sub)
				   (lambda ()
					 (let ([i (unbox count)])
					   (if (> i 1)
						   (set-box! count (sub1 i))
						   (on-completed sub))))
				   sub))
	  lst))))


(define/contract (rx/concat* lst)
  (-> (listof observable?) observable?)
  (foldl (lambda (obs2 obs1)
		   (create-observable
			(lambda (sub)
			  (subscribe obs1 (funn sub) (fune sub)
						 (lambda ()
						   (subscribe obs2
									  (funn sub) (fune sub) (func sub)
									  sub))))))
		 (car lst) (cdr lst)))

(define/contract (rx/do-next obs proc)
  (-> observable? (-> any/c any) observable?)
  (create-observable
   (lambda (sub)
	 (subscribe obs (lambda (x) (proc x) (on-next sub x)) (fune sub) (func sub)))))

(define/contract (rx/do-error obs proc)
  (-> observable? (-> error? any) observable?)
  (create-observable
   (lambda (sub)
	 (subscribe obs	(funn sub) (lambda (e) (proc e) (on-error sub e)) (func sub)))))

(define/contract (rx/do-completed obs proc)
  (-> observable? (-> any) observable?)
  (create-observable
   (lambda (sub)
	 (subscribe obs (funn sub) (fune sub) (lambda () (proc) (on-completed sub))))))

(define/contract (rx/delay obs second)
  (-> observable? real? observable?)
  (create-observable
   (lambda (sub)
	 (thread (lambda ()
			   (sleep second)
			   (subscribe obs (funn sub) (fune sub) (func sub)))))))

;; error
(define/contract (rx/catch obs proc)
  (-> observable? (-> error? observable?) observable?)
  (create-observable
   (lambda (sub)
	 (subscribe obs
				(funn sub)
				(lambda (e) (subscribe (proc e) (funn sub) (fune sub) (func sub) sub))
				(func sub)))))

;; map
(define/contract (rx/map obs proc)
  (-> observable? (-> any/c any/c) observable?)
  (create-observable
   (lambda (sub)
	 (subscribe obs (lambda (x) (on-next sub (proc x))) (fune sub) (func sub)))))

(define/contract (rx/flat-map obs proc)
  (-> observable? (-> any/c observable?) observable?)
  (create-observable
   (lambda (sub)
	 (define count (box 1))
	 (subscribe obs
				(lambda (x)
				  (set-box! count (add1 (unbox count)))
				  (subscribe (proc x) (funn sub) (fune sub)
							 (lambda () (set-box! count (sub1 (unbox count)))
								 (when (= (unbox count) 0)
								   (on-completed sub)))
							 sub))
				(fune sub)
				(lambda () (set-box! count (sub1 (unbox count)))
					(when (= (unbox count) 0)
					  (on-completed sub)))))))

;; filtering
(define/contract (rx/filter obs proc)
  (-> observable? (-> any/c any/c) observable?)
  (create-observable
   (lambda (sub)
	 (subscribe obs (lambda (x) (when (proc x) (on-next sub x)))
				(fune sub) (func sub)))))

(define/contract (rx/skip obs n)
  (-> observable? exact-nonnegative-integer? observable?)
  (create-observable
   (lambda (sub)
	 (define count (box 1))
	 (subscribe obs
				(lambda (x)
				  (if (> (unbox count) n)
					  (on-next sub x)
					  (set-box! count (add1 (unbox count)))))
				(fune sub) (func sub)))))

(define/contract (rx/take obs n)
  (-> observable? exact-nonnegative-integer? observable?)
  (create-observable
   (lambda (sub)
	 (define count (box 0))
	 (subscribe obs
				(lambda (x)
				  (if (< (unbox count) n)
					  (begin
						(on-next sub x)
						(set-box! count (add1 (unbox count)))
						(when (>= (unbox count) n)
							(on-completed sub)))
					  (on-completed sub)))
				(fune sub) (func sub)))))

(define/contract (rx/debounce obs second)
  (-> observable? real? observable?)
  (create-observable
   (lambda (sub)
	 (define b (box #t))
	 (subscribe (rx/interval second) (lambda (x) (set-box! b #t)) void void sub)
	 (subscribe obs (lambda (x)
					  (when (unbox b)
						(set-box! b #f)
						(on-next sub x)))
				(fune sub) (func sub)))))

;; combining
(define/contract (rx/start-with obs item)
  (-> observable? any/c observable?)
  (create-observable
   (lambda (sub)
	 (on-next sub item)
	 (subscribe obs (funn sub) (fune sub) (func sub)))))

(define/contract (rx/merge obs . rest)
  (->* (observable?) () #:rest (listof observable?) observable?)
  (rx/merge* (cons obs rest)))

(define/contract (rx/concat obs . rest)
  (->* (observable?) () #:rest (listof observable?) observable?)
  (rx/concat* (cons obs rest)))

